package com.lvmama.task.job.thread;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.util.ConcurrentHashSet;

import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by IntelliJ IDEA.<p/>
 * User: troy-kou<p/>
 * Date: 14-5-26<p/>
 * Time: 下午3:03<p/>
 * Email:kouhongyu@163.com<p/>
 */
public class TaskThreadPoolFactory {

    private static Log log = LogFactory.getLog(TaskThreadPoolFactory.class);

    private static TaskThreadPoolFactory poolFactory;
    private ExecutorService executorService;

    /**
     * 正在运行的Task
     */
    private Set<Long> runningTaskSet = new ConcurrentHashSet<Long>();
    /**
     * 异常结束的Task
     */
    private Set<Long> abnormalEndTaskSet = new ConcurrentHashSet<Long>();
    /**
     * 正常结束的Task
     */
    private Map<Long, Queue> successfulEndTaskMap = new ConcurrentHashMap<Long, Queue>();
    /**
     * 成功结束的Task记录的队列容量
     */
    private static final int SUCCESSFUL_END_TASK_QUEUE_CAPACITY = 150;

    private TaskThreadPoolFactory() {
        executorService = Executors.newCachedThreadPool();
    }

    public static TaskThreadPoolFactory getInstance() {
        if (poolFactory == null) {
            poolFactory = new TaskThreadPoolFactory();
        }
        return poolFactory;
    }

    /**
     * Task线程加入线程池
     *
     * @param thread
     */
    public void addThread(TaskThread thread) {
        runningTaskSet.add(thread.getTask().getTaskId());
        executorService.submit(thread);
    }

    /**
     * 判断Task是否正在运行
     *
     * @param taskId
     * @return
     */
    public boolean isRunningTask(Long taskId) {
        return runningTaskSet.contains(taskId);
    }

    /**
     * 移除正在运行的Task
     *
     * @param taskId
     */
    public void removeRunningTask(Long taskId) {
        runningTaskSet.remove(taskId);
    }

    /**
     * 判断Task是否在成功结束队列中
     *
     * @param taskId
     * @param nextRunTime
     * @return
     */
    public boolean isSuccessfulEndTask(Long taskId, Long nextRunTime) {
        return successfulEndTaskMap.get(taskId) != null && successfulEndTaskMap.get(taskId).contains(nextRunTime);
    }

    /**
     * 记录成功结束的Task
     *
     * @param taskId
     * @param runTime
     */
    public void putSuccessfulEndTask(Long taskId, Long runTime) {
        if (successfulEndTaskMap.get(taskId) == null) {
            Queue<Long> queue = new ArrayBlockingQueue<Long>(SUCCESSFUL_END_TASK_QUEUE_CAPACITY);
            queue.add(runTime);
            successfulEndTaskMap.put(taskId, queue);
        } else {
            Queue queue = successfulEndTaskMap.get(taskId);
            if (!queue.offer(runTime)) {
                queue.poll();
                queue.offer(runTime);
            }
        }
    }

    /**
     * Task是否异常结束
     *
     * @param taskId
     * @return
     */
    public boolean isAbnormalEndTask(Long taskId) {
        return abnormalEndTaskSet.contains(taskId);
    }

    /**
     * 记录异常结束的Task
     *
     * @param taskId
     */
    public void putAbnormalEndTask(Long taskId) {
        abnormalEndTaskSet.add(taskId);
    }

    /**
     * 移除异常结束的Task
     *
     * @param taskId
     */
    public void removeAbnormalEndTask(Long taskId) {
        abnormalEndTaskSet.remove(taskId);
    }

}
